package com.smsc.headend.task.engine.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Lists;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.enums.errorcode.AssetErrorCode;
import com.smsc.headend.common.enums.errorcode.TaskHandlerErrorCode;
import com.smsc.headend.common.utils.*;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.collection.dto.CaptureObject;
import com.smsc.headend.module.data.dmo.GroupSubDataItem;
import com.smsc.headend.module.data.dto.*;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.entity.ScheduleReadingEntity;
import com.smsc.headend.module.data.enums.CollectionSource;
import com.smsc.headend.module.event.entity.RdEventMeter;
import com.smsc.headend.module.event.entity.RdEventMeterData;
import com.smsc.headend.module.protocol.cosem.bo.CosemData;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.protocol.cosem.util.CosemXmlUtils;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.TaskEventData;
import com.smsc.headend.module.task.dto.TaskReadData;
import com.smsc.headend.module.task.proto.TaskEventDataProto;
import com.smsc.headend.module.task.proto.TaskReadDataProto;
import com.smsc.headend.task.engine.dao.RdMeterReadsLpDAO;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.mapper.RdMeterReadsLpMapper;
import com.smsc.headend.task.engine.po.ScheduleReadingDataBo;
import com.smsc.headend.task.engine.po.ScheduleReadingPageBo;
import com.smsc.headend.task.engine.service.*;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.collection.CaptureObjsStatusService;
import com.smsc.headend.task.engine.service.dds.DdsService;
import com.smsc.headend.task.engine.utils.ReportTimeUtils;
import gurux.dlms.GXDLMSTranslator;
import gurux.dlms.TranslatorOutputType;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.assertj.core.util.Preconditions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.nio.charset.Charset;
import java.time.Instant;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DataProcessServiceImpl extends ServiceImpl<RdMeterReadsLpMapper, RdMeterReadsLp> implements DataProcessService {

    @Autowired
    @Lazy
    DataProcessService selfService;
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    CosemXmlService cosemXmlService;
    @Autowired
    AssetService assetService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    CaptureObjsStatusService captureObjsStatusService;
    @Value("${collection.needCapture:true}")
    String needCapture;
    @Value("${spring.datasource.dbnames}")
    String[] dnNames;
    @Autowired
    DdsService ddsService;
    @Autowired
    RdMeterReadsLpDAO rdMeterReadsLpDAO;

    @Autowired
    DataConvertService dataConvertService;
    @Autowired
    DeviceStatusService deviceStatusService;
    @Autowired
    DataProcessService dataProcessService;
    @Autowired
    FronzenDataService fronzenDataService;

    @Value("${spring.datasource.dbtype:oracle}")
    String dbType;

    @Autowired
    ThreadPoolExecutor saveDbExecutor;


    LoadingCache<String, List<DataItemDTO>> loadProfileCaptureObjects = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(65535)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(20)
            // expireAfterWrite设置写缓存后30秒钟过期
            .expireAfterWrite(30, TimeUnit.SECONDS)
            .build(new CacheLoader<String, List<DataItemDTO>>() {
                @Override
                public List<DataItemDTO> load(String modelAndTemplate) throws Exception {
                    String[] info = modelAndTemplate.split(":");
                    Long modelId = Long.parseLong(info[0]);
                    Long lpTemplateId = Long.parseLong(info[1]);
                    return assetManagementService.getProfileDataItemForDeviceByLpTemplateIdAndModel(modelId, lpTemplateId);
                }
            });

    @Override
    public int saveTaskReadData(TaskReadData result, CollectionSource collectionSource) {
        if (result == null || CollectionUtils.isEmpty(result.getTaskMeterReads())) {
            return 0;
        }
        log.debug("start insert {}", result.getTaskMeterReads().size());
        Long startTime = SystemClock.now();
        Long updateTime = startTime / 1000;
        result.getTaskMeterReads().forEach(meterRead -> meterRead.setTvUpdate(updateTime));
        List<RdMeterReadsLp> readsLpList = result.getTaskMeterReads().stream().distinct().collect(Collectors.toList());
        //使用kafka消息队列提交
        Long meterId = readsLpList.get(0).getMeterId();
        TaskReadData readData = new TaskReadData();
        readData.setTaskMeterReads(readsLpList);
        kafkaTemplate.send(TaskKafkaTopic.DEVICE_READ_TRANSFER, Convert.toStr(meterId), ProtoBufferUtil.toProtoBuffer(readData, TaskReadDataProto.TaskReadData.class).toByteArray());
        return 1;
    }

//    public IPage<RdMeterReadsLp> getMeterReadPage(Page<RdMeterReadsLp> page, Long meterId, String startTime, String endTime) {
//        if (StringUtils.isEmpty(startTime) || StringUtils.isEmpty(endTime)) {
//            return null;
//        }
//        QueryWrapper<RdMeterReadsLp> queryWrapper = new QueryWrapper<RdMeterReadsLp>();
//
//        queryWrapper.between("TV", Long.parseLong(startTime), Long.parseLong(endTime));
//        if (meterId != null) {
//            log.error("not support yet");
//            return null;
////            queryWrapper.eq("METER_ID", meterId);
//        }
//        queryWrapper.select("METER_ID", "DATA_ITEM_ID", "TV", "VALUE", "VALIDITY", "SRC", "LP_DATA_ITEM_ID");
//        Page result = getBaseMapper().selectPage(page, queryWrapper);
////        result.setRecords(getBaseMapper().getPageData(startTime, endTime, page.getCurrent() * page.getSize(), (page.getCurrent() - 1) * page.getSize()));
//        if ("postgresql".equals(dbType)) {
//            result.setRecords(getBaseMapper().getPageDataForPgsql(Long.parseLong(startTime), Long.parseLong(endTime), page.getCurrent() * page.getSize(), (page.getCurrent() - 1) * page.getSize()));
//        } else if ("oracle".equals(dbType)) {
//            result.setRecords(getBaseMapper().getPageData(startTime, endTime, page.getCurrent() * page.getSize(), (page.getCurrent() - 1) * page.getSize()));
//        }
//
//        return result;
//    }

    @Override
    public List<DataItemDTO> getLoadProfileCaptureObjects(Long lpTemplateId, Long lpDataItemId, Long meterId) {
        if (lpTemplateId == null) {
            return assetManagementService.getProfileDataItemForDeviceByLpDataItem(meterId, lpDataItemId);
        }
        Meter meter = assetService.getMeterById(meterId);
        Preconditions.checkNotNull(meter);
        String key = meter.getModelId() + ":" + lpTemplateId;
        List<DataItemDTO> dataItemDTOS = null;
        try {
            dataItemDTOS = loadProfileCaptureObjects.get(key);
        } catch (Exception e) {
            log.error("get capture objects failed", e);
        }
        return dataItemDTOS;
    }

    @Override
    public Boolean saveLoadProfileRecord(List<RdMeterReadsLp> readsLps) throws ExecutionException, InterruptedException {
        long start = SystemClock.now();
        Map<Long, List<RdMeterReadsLp>> dbShardRecords = readsLps.stream().collect(Collectors.groupingBy(rd -> rd.getMeterId() % dnNames.length));
        List<? extends Future<?>> collect = dbShardRecords.entrySet().stream().map(kv -> saveDbExecutor.submit(() -> {
            selfService.saveShardedByMeterLp(kv.getValue(), kv.getKey());
        })).collect(Collectors.toList());
        for (Future<?> future : collect) {
            future.get();
        }
        log.debug("savedbShardRecords:size:{},cost:{}ms", dbShardRecords.size(), SystemClock.now() - start);
        return true;
    }

    @Override
    public R createExportLoadProfileTask(LpDdsDTO lpDdsDto) {
        List<Long> meterIds = lpDdsDto.getMeterIds();
        if (null == meterIds || 0 == meterIds.size()) {
            return R.error(AssetErrorCode.METER_ID_MISSING.getErrorCode(), AssetErrorCode.METER_ID_MISSING.getErrorText());
        }
//        List<RdMeterReadsLp> rdMeterReadsLps = getBaseMapper().getList(lpDdsDto.getStartTime(), lpDdsDto.getEndTime(), lpDdsDto.getLpDataItemId(), meterIds);
//        TaskReadData readData = new TaskReadData();
//        readData.setTaskMeterReads(rdMeterReadsLps);
//        List<LpDdsDTO.MeterId> meterIdList = new ArrayList<>();
//        for(Long meterId:meterIds) {
//            LpDdsDTO.MeterId meterId1 = new LpDdsDTO.MeterId();
//            meterId1.setMeterId(meterId);
//            meterIdList.add(meterId1);
//        }
//        lpDdsDto.setMeterIdList(meterIdList);
        kafkaTemplate.send(TaskKafkaTopic.EXPORT_LP_DATA, JSONUtil.toJsonStr(lpDdsDto).getBytes(Charset.forName("utf-8")));
        return R.ok();
    }

    @Override
    public void exportLoadProfileToDds(LpDdsDTO lpDdsDto) {
        List<Long> meterIds = lpDdsDto.getMeterIds();
        if (null == meterIds || 0 == meterIds.size()) {
            return;
        }
        List<RdMeterReadsLp> rdMeterReadsLps = getBaseMapper().getList(lpDdsDto.getStartTime(), lpDdsDto.getEndTime(), lpDdsDto.getLpDataItemId(), meterIds);
        ddsService.transformMeterReadResult(rdMeterReadsLps);
    }

    @Override
    public IPage<RdMeterReadsLp> pageRdMeterReadsLp(Long lpDataItemId, Long meterId, Long startTv, Long endTv, Integer page, Integer pageSize) {
        Page pa = new Page();
        pa.setCurrent(page);
        pa.setSize(pageSize);
        IPage<RdMeterReadsLp> pageRdMeterReadsLp = this.getBaseMapper().getPageRdMeterReadsLp(pa, meterId, lpDataItemId, startTv, endTv);
        return pageRdMeterReadsLp;
    }

    @Override
    public IPage<RdMeterReadsLp> pageRdMeterReadsLpByRange(Long lpDataItemId, List<Long> meterIds, Long startTv, Long endTv, Integer page, Integer pageSize) {
        Page pa = new Page();
        pa.setCurrent(page);
        pa.setSize(pageSize);
        IPage<RdMeterReadsLp> pageRdMeterReadsLp = this.getBaseMapper().getPageRdMeterReadsLpByIds(pa, meterIds, lpDataItemId, startTv, endTv);
        return pageRdMeterReadsLp;
    }


    public Integer countByTvs(Long lpDataItemId, List<Long> meterIds, Long startTv, Long endTv) {
        return this.getBaseMapper().countMetersByTv(lpDataItemId, meterIds, startTv, endTv);
    }

    @Override
    public ScheduleReadingDataBo lpDataPivotByOffsetAndLimit(Long[] meterIds, Long lpDataItemId, Long[] subDataItemIds, Long startTv, Long endTv, Integer offset, Integer limit) {
        List<Date> dateList = null;
        if (lpDataItemId == 1020070002) {
            // 日的最大按五天分割
            dateList = ReportTimeUtils.cutDate("D", startTv, endTv, 5);
        } else if (lpDataItemId == 1020070001) {
            // 月数据按一个月计算
            dateList = ReportTimeUtils.cutDate("M", startTv, endTv, 1);
        } else {
            // 分钟的暂时按24小时分割计算 后面再特殊处理
            dateList = ReportTimeUtils.cutDate("H", startTv, endTv, 24);
        }
        try {
            if (CollUtil.isNotEmpty(dateList)) {
                log.info("meterSize {}", meterIds.length);
                List<List<Long>> mIds =  Lists.partition(Arrays.asList(meterIds), 900);
                Set<Long> dataItemIdSet = new HashSet<>();
                List<ScheduleReadingEntity> totalSchReadingEntities = new ArrayList<>();
                for (int i = 0; i < dateList.size() -1; i++) {
                    log.info("batch {}",i);
                    for (List<Long> mList: mIds) {
                    log.info("mIds {}",mIds.size());
                        Long[] subMeterIds = ArrayUtil.toArray(mList, Long.class);
                        Long start = SystemClock.now();
                        List<RdMeterReadsLp> list = new ArrayList<>();
                        if (dbType.equals("oracle")) {
                            list = this.getBaseMapper().getLimitList(lpDataItemId, Lists.newArrayList(subMeterIds), Lists.newArrayList(subDataItemIds), dateList.get(i).getTime() / 1000, dateList.get(i+1).getTime() / 1000, offset, offset + limit);
                        } else if (dbType.equals("postgresql")) {
                            list = this.getBaseMapper().getLimitListForPgsql(lpDataItemId, Lists.newArrayList(subMeterIds), Lists.newArrayList(subDataItemIds), startTv, endTv, offset, offset + limit);
                        } else {
                            log.error("lpDataPivotByOffsetAndLimit unknown dbType");
                        }
                        Long query = SystemClock.now();
                        Map<String, List<RdMeterReadsLp>> collect = list.stream().collect(Collectors.groupingBy(lp -> lp.getTv() + ":" + lp.getMeterId()));

                        List<ScheduleReadingEntity> scheduleReadingEntities = collect.entrySet().stream().map(en -> {
                            List<RdMeterReadsLp> value = en.getValue();
                            Long tv = value.get(0).getTv();
                            Long meterId = value.get(0).getMeterId();
                            Meter meter = assetService.getMeterById(meterId);
                            if (meter == null) {
                                log.info("the meter is not exist {}", meterId);
                            }
                            ScheduleReadingEntity scheduleReadingEntity = new ScheduleReadingEntity();
                            scheduleReadingEntity.put("tv", BigDecimal.valueOf(tv));
                            scheduleReadingEntity.put("src", BigDecimal.valueOf(value.get(0).getSrc()));
                            scheduleReadingEntity.put("tvUpdate", BigDecimal.valueOf(value.get(0).getTvUpdate()));
                            scheduleReadingEntity.put("lpDataItemId", lpDataItemId);
                            scheduleReadingEntity.put("meterId", meterId);
                            scheduleReadingEntity.put("mfgSn", Optional.ofNullable(meter).orElse(new Meter()).getMfgSn());
                            scheduleReadingEntity.put("customerSn", Optional.ofNullable(meter).orElse(new Meter()).getCustomerSn());
                            scheduleReadingEntity.put("validity", value.get(0).getValidity());
                            value.stream().forEach(t -> {
                                if (!dataItemIdSet.contains(t.getDataItemId())) {
                                    dataItemIdSet.add(t.getDataItemId());
                                }

                                scheduleReadingEntity.put(Convert.toStr(t.getDataItemId()), dataConvertService.getGUIValue(t.getValue(), meter == null ? 0:meter.getModelId(), t.getDataItemId()));
                            });
                            return scheduleReadingEntity;
                        }).collect(Collectors.toList());

                        totalSchReadingEntities.addAll(scheduleReadingEntities);

                        Long now = SystemClock.now();
                        log.info("getGroupMeterData:meterSize:{}, offset:{},limit:{},query:{}ms,compose{}ms", meterIds.length, offset, limit, query - start, now - query);
                    }
                }
//                totalSchReadingEntities.sort(Comparator.comparing(ScheduleReadingEntity::getMfgSn));
                return new ScheduleReadingDataBo(dataItemIdSet, totalSchReadingEntities);
            }
        } catch (Exception e) {
            log.info("exception", e);
            return null;
        }
        return null;
    }


    @Override
    public ScheduleReadingPageBo pagePivotRdMeterReadsLp(Long lpDataItemId, Long meterId, Long startTv, Long endTv, Integer page, Integer pageSize, Long[] subDataItemIds) {
        IPage<ScheduleReadingEntity> pageDto = new Page<>();
        long startTime = SystemClock.now();
        long count = this.getBaseMapper().countByTv(lpDataItemId, meterId, startTv, endTv);
        log.info("count: {}", count);
        Integer totalPage = Convert.toInt(Math.ceil(((double) count) / pageSize));
        pageDto.setTotal(count);
        pageDto.setCurrent(page);
        pageDto.setSize(pageSize);
        pageDto.setPages(totalPage);
        if (page > totalPage || page <= 0) {
            return new ScheduleReadingPageBo(Collections.emptySet(), new PageUtils(pageDto));
        }
        Date startDate = new Date(startTv * 1000);
        Date endDate = new Date(endTv * 1000);
        Date endMonthFstDate = DateUtils.truncate(endDate, Calendar.MONTH);
        int endRow = page * pageSize;
        int startRow = (page - 1) * pageSize;
        List<Long> list = new ArrayList();
        //Same Month
        if (!startDate.before(endMonthFstDate)) {
            log.info("1Month");
            if ("oracle".equals(dbType)) {
                list = this.getBaseMapper().getPagedTv(lpDataItemId, meterId, startTv, endTv, startRow, endRow);
            } else if ("postgresql".equals(dbType)) {
                list = this.getBaseMapper().getPagedTvForPgsql(lpDataItemId, meterId, startTv, endTv, startRow, endRow);
            } else {
                log.error("unknown db type");
            }
        }
        //Cross two month
        else {
            log.info("2Month");
            Integer fstMonthCount = this.getBaseMapper().countByTv(lpDataItemId, meterId, startTv, endMonthFstDate.getTime() / 1000 - 1);
            log.info("1stMonthCount: {}", fstMonthCount);
            if (startRow > fstMonthCount) {
                log.info("query in 2ndMonth");
                if ("oracle".equals(dbType)) {
                    list = this.getBaseMapper().getPagedTv(lpDataItemId, meterId, endMonthFstDate.getTime() / 1000, endTv, startRow - fstMonthCount, endRow - fstMonthCount);
                } else if ("postgresql".equals(dbType)) {
                    list = this.getBaseMapper().getPagedTvForPgsql(lpDataItemId, meterId, endMonthFstDate.getTime() / 1000, endTv, startRow - fstMonthCount, endRow - fstMonthCount);
                } else {
                    log.error("unknown db type");
                }
            } else if (fstMonthCount >= endRow) {
                log.info("query in 1stMonth");
                if ("oracle".equals(dbType)) {
                    list = this.getBaseMapper().getPagedTv(lpDataItemId, meterId, startTv, endMonthFstDate.getTime() / 1000 - 1, startRow, endRow);
                } else if ("postgresql".equals(dbType)) {
                    list = this.getBaseMapper().getPagedTvForPgsql(lpDataItemId, meterId, startTv, endMonthFstDate.getTime() / 1000 - 1, startRow, endRow);
                } else {
                    log.error("unknown db type");
                }
            } else {
                List<Long> tvListFstMonth = new ArrayList<>();
                List<Long> tvListLstMonth = new ArrayList<>();
                if ("oracle".equals(dbType)) {
                    tvListFstMonth = this.getBaseMapper().getPagedTv(lpDataItemId, meterId, startTv, endMonthFstDate.getTime() / 1000 - 1, startRow, fstMonthCount);
                    tvListLstMonth = this.getBaseMapper().getPagedTv(lpDataItemId, meterId, endMonthFstDate.getTime() / 1000, endTv, 0, endRow - fstMonthCount);
                } else if ("postgresql".equals(dbType)) {
                    tvListFstMonth = this.getBaseMapper().getPagedTvForPgsql(lpDataItemId, meterId, startTv, endMonthFstDate.getTime() / 1000 - 1, startRow, fstMonthCount);
                    tvListLstMonth = this.getBaseMapper().getPagedTvForPgsql(lpDataItemId, meterId, endMonthFstDate.getTime() / 1000, endTv, 0, endRow - fstMonthCount);
                } else {
                    log.error("unknown db type");
                }
                list.addAll(tvListLstMonth);
                list.addAll(tvListFstMonth);
            }
        }
        log.info("tvSize: {}", list.size());
        List<Long> collect = list.stream().sorted().limit(pageSize).collect(Collectors.toList());
        Long start = collect.stream().reduce(Long::min).get();
        Long end = collect.stream().reduce(Long::max).get();
        log.info("page: {} ,start:{}, end:{}", page, Instant.ofEpochSecond(start).atZone(ZoneId.systemDefault()), Instant.ofEpochSecond(end).atZone(ZoneId.systemDefault()));
        List<RdMeterReadsLp> pageList = this.getBaseMapper().getPageList(lpDataItemId, meterId, subDataItemIds == null ? null : Lists.newArrayList(subDataItemIds), start, end);
        long queryEndTime = SystemClock.now();
        log.info("cost: {}ms , pageList: {}", queryEndTime - startTime, pageList.size());
        Map<Long, List<RdMeterReadsLp>> tvMap = pageList.stream().collect(Collectors.groupingBy(o -> o.getTv()));
        Meter meter = assetService.getMeterById(meterId);
        Set<Long> dataItemIdSet = new HashSet<>();
        List<ScheduleReadingEntity> resultEntities = tvMap.entrySet().stream().map(p -> {
            List<RdMeterReadsLp> value = p.getValue();
            ScheduleReadingEntity scheduleReadingEntity = new ScheduleReadingEntity();
            scheduleReadingEntity.put("tv", BigDecimal.valueOf(p.getKey()));
            scheduleReadingEntity.put("src", BigDecimal.valueOf(value.get(0).getSrc()));
            scheduleReadingEntity.put("tvUpdate", BigDecimal.valueOf(value.get(0).getTvUpdate()));
            scheduleReadingEntity.put("lpDataItemId", lpDataItemId);
            scheduleReadingEntity.put("meterId", meterId);
            scheduleReadingEntity.put("mfgSn", Optional.ofNullable(meter).orElse(new Meter()).getMfgSn());
            scheduleReadingEntity.put("customerSn", Optional.ofNullable(meter).orElse(new Meter()).getCustomerSn());
            scheduleReadingEntity.put("validity", value.get(0).getValidity());
            value.stream().forEach(t -> {
                if (!dataItemIdSet.contains(t.getDataItemId())) {
                    dataItemIdSet.add(t.getDataItemId());
                }
                scheduleReadingEntity.put(Convert.toStr(t.getDataItemId()), dataConvertService.getGUIValue(t.getValue(), meter.getModelId(), t.getDataItemId()));
            });
            return scheduleReadingEntity;
        }).collect(Collectors.toList());
        List<ScheduleReadingEntity> finalList = resultEntities.stream().sorted(Comparator.comparing(o -> Convert.toStr(o.get("tv")))).collect(Collectors.toList());
        pageDto.setRecords(finalList);
        log.info("compose cost: {}ms ", SystemClock.now() - queryEndTime);
        return new ScheduleReadingPageBo(dataItemIdSet, new PageUtils(pageDto));
    }

    public Integer saveShardedByMeterLp(List<RdMeterReadsLp> rdMeterReadsLps, Long key) {
        long start = SystemClock.now();
        Map<String, List<RdMeterReadsLp>> groupMap = rdMeterReadsLps.stream().collect(Collectors.groupingBy(rd -> {
                    Date time = new Date(rd.getTv() * 1000);
                    return DateFormatUtils.format(time, "yyyyMM");
                }
        ));
        for (Map.Entry<String, List<RdMeterReadsLp>> entry : groupMap.entrySet()) {
            String yearMonth = entry.getKey();
            log.info("save to:{}", dnNames[key.intValue()] + ".RD_METER_READS_LP_" + yearMonth);
            if ("oracle".equals(dbType)) {
                rdMeterReadsLpDAO.batchSaveRdMeterReadsLp(groupMap.get(yearMonth), dnNames[key.intValue()], yearMonth);
            } else if ("postgresql".equals(dbType)) {
                rdMeterReadsLpDAO.batchSaveRdMeterReadsLpForPgsql(groupMap.get(yearMonth), dnNames[key.intValue()], yearMonth);
            } else {
                log.error("unknown db type");
            }

        }
        log.warn("saveShardedCost：{}ms", SystemClock.now() - start);
        return 1;
    }

    @Override
    public void saveDLMSGatewayFtpCollectionData(FtpCollectionDTO result) {
        Map<String, List<FtpCollectionDTO.RecordWithPacket>> meterSnRecordMap = result.getRecords().stream().collect(Collectors.groupingBy(FtpCollectionDTO.RecordWithPacket::getMeterSn));
        for (Map.Entry<String, List<FtpCollectionDTO.RecordWithPacket>> meterSn : meterSnRecordMap.entrySet()) {
            saveMeterCollection(meterSn.getKey(), meterSn.getValue());
        }
    }

    private void saveMeterCollection(String meterSn, List<FtpCollectionDTO.RecordWithPacket> recordWithPacketList) {
        Meter meter = assetManagementService.getMeterByMfgSn(meterSn);
        if (meter == null) {
            log.error("ftp collect:  meter {} not found", meterSn);
            return;
        }
        Map<String, List<FtpCollectionDTO.RecordWithPacket>> obisRecordMap = recordWithPacketList.stream().collect(Collectors.groupingBy(FtpCollectionDTO.RecordWithPacket::getObis));
        for (Map.Entry<String, List<FtpCollectionDTO.RecordWithPacket>> obisEntry : obisRecordMap.entrySet()) {
            String obis = obisEntry.getKey();
            List<FtpCollectionDTO.RecordWithPacket> recordList = obisEntry.getValue();
            Long dataItem = assetManagementService.getLpDataItemByObis(meter.getMeterId(), obis);
            if (dataItem == null) {
                log.error("ftp collect: no lpDataItem, meterId={}, obis={}", meter.getMeterId(), obis);
                continue;
            }

            //FTP定抄捕获对象入库
            List<DataItemDTO> captureList = assetManagementService.getProfileDataItemForDeviceByLpDataItem(meter.getMeterId(), dataItem);
            if (CollectionUtils.isEmpty(captureList)) {
                log.error("ftp collect: Meter {} profile in obis {} have no capture list setting", meterSn, obis);
            } else {
                Long lpTemplateId = assetManagementService.getLpTemplateIdByDataItemId(meter.getMeterId(), dataItem);
                List<RdMeterReadsLp> rdMeterReadsLps = recordList.stream().map(o -> buildRdMeterReadRecord(o, lpTemplateId, dataItem, captureList, meter)).filter(o -> CollectionUtils.isNotEmpty(o)).flatMap(Collection::stream).collect(Collectors.toList());
                TaskReadData taskReadData = new TaskReadData();
                taskReadData.setIsODR(false);
                taskReadData.setTaskMeterReads(rdMeterReadsLps);
                this.saveTaskReadData(taskReadData, CollectionSource.FTP);
                deviceStatusService.updateMeterReadDataStatus(meter);
            }

            //FTP定抄事件入库
            List<DataItemDTO> logBookList = assetManagementService.getEventProfileDataItemByLbDataItemId(meter.getMeterId(), dataItem, DeviceType.METER.getId());
            if (CollectionUtils.isEmpty(logBookList)) {
                log.error("ftp collect: Meter {} logbook in obis {} have no logbook list setting", meterSn, obis);
            } else {
                List<RdEventMeter> rdEventMeters = recordList.stream().map(o -> buildRdEventMeterData(o, dataItem, logBookList, meter)).filter(o -> CollectionUtils.isNotEmpty(o)).flatMap(Collection::stream).collect(Collectors.toList());
                log.debug("saveMeterCollection rdEventMeters={}", JSONUtil.toJsonStr(rdEventMeters));
                TaskEventData readData = new TaskEventData();
                readData.setIsODR(false);
                readData.setRdEventMeter(rdEventMeters);
                kafkaTemplate.send(TaskKafkaTopic.DEVICE_EVENT_TRANSFER, ProtoBufferUtil.toProtoBuffer(readData, TaskEventDataProto.TaskEventData.class).toByteArray());
            }
        }
    }

    private List<RdMeterReadsLp> buildRdMeterReadRecord(FtpCollectionDTO.RecordWithPacket recordWithPacket, Long lpTemplateId, Long lpDataItemId, List<DataItemDTO> dataItemCosemList, Meter meter) {
        String dlmsPacketString = recordWithPacket.getMessage();
        GXDLMSTranslator translator = new GXDLMSTranslator(TranslatorOutputType.SIMPLE_XML);
        try {
            String xml = translator.pduToXml(ByteUtils.hexStringToByte("C40141" + dlmsPacketString));
            CosemData cosemData = CosemXmlUtils.translateCosemXml(xml);
            CosemData dataNodeCosemData = CosemXmlUtils.getDataFromPduCosemData(cosemData);
            return getRdMeterReadLpList(dataNodeCosemData, lpTemplateId, lpDataItemId, dataItemCosemList, meter, CollectionSource.FTP);
        } catch (Exception | UDISTaskHandlerException e) {
            log.error("translate gateway dlms collection data of load profile fail, {}", recordWithPacket.getMessage(), e);
        }
        return null;
    }

    @Override
    public List<RdMeterReadsLp> getRdMeterReadLpList(CosemData cosemData, Long lpTemplateId, Long lpDataItemId, List<DataItemDTO> dataItemCosemList, Meter meter, CollectionSource collectionSource) throws UDISTaskHandlerException {
        List<RdMeterReadsLp> meterReadsCol = new ArrayList<>();
        //ftp的定抄数据不比较捕获清单
        boolean needCp = (collectionSource != CollectionSource.FTP && collectionSource != CollectionSource.PUSH) ? Convert.toBool(needCapture) : false;
        List<DataItemDTO> actualCaptureObjs = mergeCaptureObjs(meter.getMeterId(), lpTemplateId, dataItemCosemList, cosemData.getChild().get(0).getChild().size(), needCp);
        if (CollUtil.isEmpty(actualCaptureObjs)) {
            log.error("No Capture objs: meterId={}, lpTemplateId={}, lpDataItemId={}, dataItemList={}", meter.getMeterId(), lpTemplateId, lpDataItemId, JSONUtil.toJsonStr(dataItemCosemList));
            throw new UDISTaskHandlerException(TaskHandlerErrorCode.PROFILE_CAPTURE_OBJECTS_SIZE_MISMATCH.toString());
        }

        try {
            List<DataItemDTO> finalActualCaptureObjs = actualCaptureObjs;
            cosemData.getChild().stream().forEach(structData -> {
                List<RdMeterReadsLp> oneTimeMeterReads = new ArrayList<>();
                Integer structureLen = structData.getChild().size();
                log.debug("structure len={} cosemData={}", structureLen, structData);
                String timeString = null;
                Integer validity = 0;
                for (DataItemDTO dataItemCosem : finalActualCaptureObjs) {
                    DLMSSubDataItemDTO dlmsSubDataItemDTO = (DLMSSubDataItemDTO) dataItemCosem;
                    if (dlmsSubDataItemDTO.getIndex() >= structData.getChild().size()) {
                        log.warn("profile object list longer than list configured in UDIS");
                        break;
                    }
                    CosemData data = structData.getChild().get(dlmsSubDataItemDTO.getIndex());
                    if (dataItemCosem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.TIME_DATE.getCode())) {
                        Object dValue = cosemXmlService.getData(data.getType(), data.getValue());
                        if (dValue instanceof Date) {
                            timeString = data.getValue();
                        } else {
                            timeString = Convert.toStr(cosemXmlService.getData(data.getType(), data.getValue()));
                        }
                        continue;
                    }
                    if (dataItemCosem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.PROFILE_STATUS.getCode())) {
                        validity = Convert.toInt(cosemXmlService.getData(data.getType(), data.getValue()));
                        continue;
                    }
                    RdMeterReadsLp meterReadsLp = getRdMeterReadLp(dataItemCosem, lpDataItemId, meter.getMeterId(), String.valueOf(cosemXmlService.getData(data.getType(), data.getValue())), collectionSource);
                    oneTimeMeterReads.add(meterReadsLp);
                }

                Long tvUpdate = SystemClock.now() / 1000;
                for (RdMeterReadsLp oneTimeMeterRead : oneTimeMeterReads) {
                    Date date = CosemUtils.parseOctStringDataTime(timeString);
                    oneTimeMeterRead.setTv(Long.valueOf(date.getTime() / 1000));
                    oneTimeMeterRead.setValidity(validity);
                    oneTimeMeterRead.setSrc(collectionSource.getSource());
                    oneTimeMeterRead.setTvUpdate(tvUpdate);
                }
                meterReadsCol.addAll(oneTimeMeterReads);
            });
        } catch (Exception e) {
            log.error("Collect persist: meterId={}, {}", meter.getMeterId(), cosemData, e);
        }

        return meterReadsCol;
    }

    @Override
    public List<RdMeterReadsLp> getAnsiRdMeterReadLpList(List<AnsiDataDTO> ansiDataList, Long lpTemplateId, Long lpDataItemId, List<DataItemDTO> dataItemAnsiList, Meter meter, CollectionSource collectionSource) {
        List<RdMeterReadsLp> meterReadsCol = new ArrayList<>();
        long now = SystemClock.now() / 1000;
        ansiDataList.stream().forEach(ansiDataDTO -> {
            for (int i = 0; i < ansiDataDTO.getList().length; i++) {
                Integer value = ansiDataDTO.getList()[i];
                DataItemDTO ansiDataItem = dataItemAnsiList.get(i);
                RdMeterReadsLp meterReadsLp = new RdMeterReadsLp();
                meterReadsLp.setDataItemId(ansiDataItem.getSubDataItemId());
                meterReadsLp.setLpDataItemId(lpDataItemId);
                meterReadsLp.setMeterId(meter.getMeterId());
                meterReadsLp.setValue(value.toString());
                meterReadsLp.setTv(ansiDataDTO.getIntervalTime());
                meterReadsLp.setValidity(0);
                meterReadsLp.setSrc(collectionSource.getSource());
                meterReadsLp.setTvUpdate(now);
                meterReadsCol.add(meterReadsLp);
            }
        });

        return meterReadsCol;
    }

    /**
     * 构造表计事件数据
     *
     * @param recordWithPacket
     * @param lbDataItemId
     * @param dataItemCosemList
     * @param meter
     * @return
     */
    private List<RdEventMeter> buildRdEventMeterData(FtpCollectionDTO.RecordWithPacket recordWithPacket, Long lbDataItemId, List<DataItemDTO> dataItemCosemList, Meter meter) {
        String dlmsPacketString = recordWithPacket.getMessage();
        GXDLMSTranslator translator = new GXDLMSTranslator(TranslatorOutputType.SIMPLE_XML);
        try {
            String xml = translator.pduToXml(ByteUtils.hexStringToByte("C40141" + dlmsPacketString));
            CosemData cosemData = CosemXmlUtils.translateCosemXml(xml);
            log.debug("buildRdEventMeterData: CosemData={}", JSONUtil.toJsonStr(cosemData));
            CosemData dataNodeCosemData = CosemXmlUtils.getDataFromPduCosemData(cosemData);
            return getRdEventMeterList(dataNodeCosemData, lbDataItemId, dataItemCosemList, meter, CollectionSource.FTP);
        } catch (Exception e) {
            log.error("translate gateway dlms collection data of meter event fail, {}", recordWithPacket.getMessage(), e);
        }
        return null;
    }

    /**
     * CosemData 赋值给List<RdEventMeter>
     *
     * @param cosemData
     * @param lbDataItemId
     * @param dataItemCosemList
     * @param meter
     * @param collectionSource
     * @return
     */
    private List<RdEventMeter> getRdEventMeterList(CosemData cosemData, Long lbDataItemId, List<DataItemDTO> dataItemCosemList, Meter meter, CollectionSource collectionSource) {
        List<RdEventMeter> rdEventMeterList = new ArrayList<>();
        cosemData.getChild().stream().forEach(structData -> {
            Integer structureLen = structData.getChild().size();
            log.debug("structure len={} cosemData={}", structureLen, JSONUtil.toJsonStr(structData));
            String timeString = null;
            Integer eventType = null;
            Integer eventCode = null;

            RdEventMeter rdEventMeter = new RdEventMeter();
            List<RdEventMeterData> rdEventMeterDataList = new ArrayList<>();
            log.debug("getRdEventMeterList lbDataItemId={}", lbDataItemId);
            for (int i = 0; i < dataItemCosemList.size(); i++) {
                CosemData data = structData.getChild().get(i);
                //1.Clock入库RD_EVENT_METER表
                if (dataItemCosemList.get(i).getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.TIME_DATE.getCode())) {
                    Object dValue = cosemXmlService.getData(data.getType(), data.getValue());
                    if (dValue instanceof Date) {
                        timeString = data.getValue();
                    } else {
                        timeString = Convert.toStr(cosemXmlService.getData(data.getType(), data.getValue()));
                    }
                    continue;
                }

                //2.EVENT_CODE入库RD_EVENT_METER表
                if (dataItemCosemList.get(i).getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.EVENT_CODE.getCode())) {
                    String dValue = cosemXmlService.getData(data.getType(), data.getValue()).toString();
                    eventCode = Integer.parseInt(dValue, 16);
                    eventType = assetManagementService.getModelEventType(meter.getModelId(), eventCode, lbDataItemId, DeviceType.METER.getId());
                    continue;
                }

                //3.其他入库RD_EVENT_METER_DATA表
                RdEventMeterData rdEventMeterData = new RdEventMeterData();
                rdEventMeterData.setSubDataItemId(dataItemCosemList.get(i).getSubDataItemId());
                String value = String.valueOf(cosemXmlService.getData(data.getType(), data.getValue()));
                if (CosemUtils.isDateTimeString(value)) {
                    value = DateFormatUtils.format(CosemUtils.parseOctStringDataTime(value), "yyyy-MM-dd HH:mm:ss");
                }
                rdEventMeterData.setValue(value);

                rdEventMeterDataList.add(rdEventMeterData);
            }
            Date date = CosemUtils.parseOctStringDataTime(timeString);
            rdEventMeter.setMeterId(meter.getMeterId());
            rdEventMeter.setLbDataItemId(lbDataItemId);
            rdEventMeter.setTv(date.getTime() / 1000);
            rdEventMeter.setTvUpdate(SystemClock.now() / 1000);
            rdEventMeter.setSubData(rdEventMeterDataList);
            rdEventMeter.setEventType(eventType);
            rdEventMeter.setEventCode(eventCode);
            rdEventMeter.setReceiveType(2);
            rdEventMeterList.add(rdEventMeter);
        });
        return rdEventMeterList;
    }

    private List<DataItemDTO> mergeCaptureObjs(Long meterId, Long lpTemplateId, List<DataItemDTO> dataItemCosemList, int dataStructureLength, boolean needCp) {
        List<CaptureObject> captureObjs;
        if (needCp) {
            captureObjs = captureObjsStatusService.getCaptureObjs(meterId, lpTemplateId);
            if (CollUtil.isEmpty(captureObjs)) {
                return null;
            }
            List<DataItemDTO> resultList = CollUtil.newArrayList();
            //取交集
            for (DataItemDTO dataItemDTO : dataItemCosemList) {
                for (CaptureObject captureObj : captureObjs) {
                    if (dataItemDTO.getSubDataItemId().equals(captureObj.getSubDataItemId())) {
                        ((DLMSSubDataItemDTO) dataItemDTO).setIndex(captureObj.getIndex());
                        resultList.add(dataItemDTO);
                        break;
                    }
                }
            }
            return resultList;
        } else {
            if (dataStructureLength != dataItemCosemList.size()) {
                log.error("collect: meterId={}, lpTemplateId={}, config={}, actual={}", meterId, lpTemplateId, dataItemCosemList.size(), dataStructureLength);
                return null;
            }
            for (int i = 0; i < dataItemCosemList.size(); i++) {
                DataItemDTO dataItemDTO = dataItemCosemList.get(i);
                ((DLMSSubDataItemDTO) dataItemDTO).setIndex(i);
            }
            return dataItemCosemList;
        }
    }

    private RdMeterReadsLp getRdMeterReadLp(DataItemDTO dataItemCosem, Long lpDataItemId, Long meterId, String value, CollectionSource collectionSource) {
        RdMeterReadsLp meterReadsLp = new RdMeterReadsLp();
        meterReadsLp.setDataItemId(dataItemCosem.getSubDataItemId());
        meterReadsLp.setLpDataItemId(lpDataItemId);
//        if (isCollectionSourceNeedName(collectionSource)) {
//            meterReadsLp.setDataItemName(((DLMSSubDataItemDTO) dataItemCosem).getDataItemName().intern());
//        }

        meterReadsLp.setMeterId(meterId);
        if (CosemUtils.isDateTimeString(value)) {
            value = DateFormatUtils.format(CosemUtils.parseOctStringDataTime(value), "yyyy-MM-dd HH:mm:ss");
        }
        meterReadsLp.setValue(value);
//        meterReadsLp.setTvUpdate(System.currentTimeMillis() / 1000);
        return meterReadsLp;
    }

    private boolean isCollectionSourceNeedName(CollectionSource collectionSource) {
        if (CollectionSource.ODR == collectionSource) {
            return true;
        }

        return false;
    }

    @Override
    public int getProfilesStatusSeq(List<DataItemDTO> dataItemCosemList) {
        int clockSeq = -1;
        boolean found = false;
        for (DataItemDTO modelDataItemCosem : dataItemCosemList) {
            clockSeq++;
            if (modelDataItemCosem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.PROFILE_STATUS.getCode())) {
                found = true;
                break;
            }
        }
        if (!found) {
            return -1;
        }
        return clockSeq;
    }


    @Override
    public int getClockIndexInConfigDataItem(List<DataItemDTO> dataItemCosemList) {
        int clockSeq = -1;
        if (CollectionUtils.isEmpty(dataItemCosemList)) {
            return -1;
        }
        boolean found = false;
        for (DataItemDTO modelDataItemCosem : dataItemCosemList) {
            clockSeq++;
            if (modelDataItemCosem.getDataGroupIds().contains(GroupSubDataItem.SpecialDataGroup.TIME_DATE.getCode())) {
                found = true;
                break;
            }
        }
        if (!found) {
            return -1;
        }
        return clockSeq;
    }

    @Override
    public List<RdMeterReadsLp> getRdMeterReadLpList(Long meterId, Long lpDataItemId, Long subDataItemId, Long startTv, Long endTv) {
        List<RdMeterReadsLp> rdMeterReadsLpList = this.baseMapper.getRdMeterReadsLp(meterId, lpDataItemId, subDataItemId, startTv, endTv);
        return rdMeterReadsLpList;
    }

    @Override
    public List<RdMeterReadsLp> getRdMeterReadLpList(List<Long> meterIds, Long subDataItemId, Long startTime, Long endTime) {
        return this.baseMapper.getListBySubDataItem(meterIds, subDataItemId, startTime, endTime);
    }


    /**
     * 从数据库获取冻结数据， 如果不存在 生成随抄任务抄读后 ; 服务于Nesco vending项目的月冻结补偿机制
     * 只支持单个表，一种类型的曲线, 一个tv
     *
     * @param lpQueryDTO
     * @return
     */
    @Override
    public List<RdMeterReadsLp> queryElectricMeterData(LpQueryDTO lpQueryDTO) {
        //Query RdMeterReads in database
        log.info("startQuery {}", lpQueryDTO);
        List<RdMeterReadsLp> frozenDataList = this.getBaseMapper().getFrozenDataList(lpQueryDTO.getMeterId(), lpQueryDTO.getLpDataItemId(), lpQueryDTO.getTv());
        log.info("queryEnd,size:{}", frozenDataList.size());
        return frozenDataList;
    }

    @Override
    public List<RdMeterReadsLp> queryElectricMeterDataByDaily(LpQueryDTO lpQueryDTO) {
        log.info("startQuery {}", lpQueryDTO);
        Long profileStartTv = lpQueryDTO.getTv() - UDateUtils.SECONDS_OF_ONE_HOUR;
        Long profileEndTv = lpQueryDTO.getTv() + UDateUtils.SECONDS_OF_ONE_HOUR;
        long endTv = (profileStartTv + profileEndTv) / 2;
        Date date = new Date(endTv * org.apache.commons.lang3.time.DateUtils.MILLIS_PER_SECOND);
        Date startDate = org.apache.commons.lang3.time.DateUtils.addMonths(date, -1);
        long startTv = startDate.getTime() / org.apache.commons.lang3.time.DateUtils.MILLIS_PER_SECOND;
        List<RdMeterReadsLp> dailyProfileData = dataProcessService.getRdMeterReadLpList(lpQueryDTO.getMeterId(), 1020070002L, null, startTv, endTv - 1);
        if (CollectionUtil.isEmpty(dailyProfileData)) {
            log.info("noDailyDataExistsInThisMonth,,meterId:{}", lpQueryDTO.getMeterId());
            return null;
        }
        log.info("dailyDataExists,size:{}", dailyProfileData.size());
        boolean dailyMonthTvExists = dailyProfileData.stream().filter(p -> p.getTv().equals(endTv)).findAny().isPresent();
        List<RdMeterReadsLp> calculatedMonthlyData = null;
        Long lpDataItemId = 1020070001L;
        try {
            if (dailyMonthTvExists) {
                calculatedMonthlyData = dailyProfileData.stream().filter(p -> p.getTv().equals(endTv)).collect(Collectors.toList());
            } else {
                calculatedMonthlyData = fronzenDataService.calculateMonthlyDataByMinMax(dailyProfileData, endTv);
            }

            Long lpTemplateId = assetManagementService.getLpTemplateIdByDataItemId(lpQueryDTO.getMeterId(), lpDataItemId);
            if (lpTemplateId == null) {
                log.info("No profile template id of the meter {}", lpQueryDTO.getMeterId());
                return null;
            }
            List<DataItemDTO> dataItemCosemList = dataProcessService.getLoadProfileCaptureObjects(lpTemplateId, lpDataItemId, lpQueryDTO.getMeterId());
            if (CollUtil.isEmpty(dataItemCosemList)) {
                log.info("No capture object list of the meter {}", lpQueryDTO.getMeterId());
                return null;
            }
            dataProcessService.getLoadProfileCaptureObjects(lpTemplateId, lpDataItemId, lpQueryDTO.getMeterId());
            calculatedMonthlyData = fronzenDataService.intersectWithCaptureObjList(calculatedMonthlyData, dataItemCosemList);
            List<String> dataItemIds = calculatedMonthlyData.stream().map(RdMeterReadsLp::getDataItemId).map(String::valueOf).collect(Collectors.toList());
            List<String> values = calculatedMonthlyData.stream().map(RdMeterReadsLp::getValue).collect(Collectors.toList());
            if (CollUtil.isNotEmpty(calculatedMonthlyData)) {
                for (RdMeterReadsLp data : calculatedMonthlyData) {
                    // 日冻结数据项计算后对应月冻结的关系(020030006102 Total energy +A  -->> 1020030000102(Current monthly consumption energy))
                    if (data.getDataItemId() == 1020030006102L) {
                        data.setDataItemId(1020030000102L);
                    }
                }
            }
            log.info("calculatedMonthData:meterId:{},dataItemId:{},value:{}", lpQueryDTO.getMeterId()
                    , String.join(",", dataItemIds), String.join(",", values));
        } catch (Exception e) {
            log.error("exception", e);
            return null;
        }
        return calculatedMonthlyData;
    }

}
